package olegroshka.src.com.bofa.olegroshka.datasource;

import java.util.Arrays;
import java.util.Iterator;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.logging.Level;
import java.util.logging.Logger;


/**
 * Joins two records (semantically similar to the left join in SQL).  
 * 
 * @author Oleg
 *
 */
public class ConcurrentJoinDataSource implements DataSource<Object[]> {

	private static class Pair<L,R> {
		private final L left;
		private final R right;
		
		public Pair(L left, R right) {
			this.left = left;
			this.right = right;
		}
		
		public L getLeft() {
			return left;
		}
		
		public R getRight() {
			return right;
		}
		
	}
	
	private class JoinedData extends Pair<Object[], Object[]> {
		public JoinedData(Object[] left, Object[] right) {
			super(left, right);
		}
	};
	
	private static final Object[] POISON = new Object[]{"POISON"};

	private final static Logger LOG = Logger.getLogger(ConcurrentJoinDataSource.class.getName());

	private final BlockingQueue<Object[]> queue = new LinkedBlockingQueue<Object[]>();
	private final ConcurrentMap<Object, JoinedData> resultMap = 
			new ConcurrentHashMap<Object, JoinedData>(1000); 
	
	private ExecutorService threadPool;
	private Future<Void> leftToRightFuture;
	private Future<Void> rightToLeftFuture;

	private final DataSource<Object[]> leftDataSource;
	private final DataSource<Object[]> rightDataSource;
	private final int leftColumn; 
	private final int rightColumn; 
	private final int leftRecordLength; 
	private final int rightRecordLength; 
	private final int joinedRecordLength; 
	
	public ConcurrentJoinDataSource(
			DataSource<Object[]> leftDataSource,
			DataSource<Object[]> rightDataSource,
			int leftColumn, int rightColumn,
			int leftRecordSize, int rightRecordSize) {
		this.leftDataSource = leftDataSource;
		this.rightDataSource = rightDataSource;
		this.leftColumn = leftColumn;
		this.rightColumn = rightColumn;
		this.leftRecordLength = leftRecordSize;
		this.rightRecordLength = rightRecordSize;
		this.joinedRecordLength = leftRecordSize + rightRecordSize;
	}

	public Iterator<Object[]> iterator() {
		
		return new Iterator<Object[]>() {
			private final AtomicReference<Object[]> currentRecord = 
					new AtomicReference<Object[]>(null);
					
			public boolean hasNext() {
				if( currentRecord.get() == null ) {
					currentRecord.set(next());
				}
				return currentRecord.get() != null;
			}

			public Object[] next() {
				Object[] joinedData = currentRecord.get();
				if( joinedData == POISON ) { 
					return null;
				}
				
				if( joinedData != null ) {
					currentRecord.set(null);
					return joinedData;
				}				
				try {
					joinedData = queue.take();
					if( joinedData == POISON ) {
						threadPool.shutdown();
						joinedData =  null;
					}
				} catch (InterruptedException e) {
					LOG.log(Level.SEVERE, "Interrupted.", e);
					Thread.currentThread().interrupt();
				}
				return joinedData;
			}

			public void remove() {
				throw new UnsupportedOperationException();
			}
		};
	}

	private void joinRows() {
		joinFromRightToLeft();
		joinFormLeftToRight();
	}

	private void joinFormLeftToRight() {
		this.leftToRightFuture = threadPool.submit(new Callable<Void>() {
			public Void call() throws Exception {
				Iterator<Object[]> iterator = leftDataSource.iterator();
				while (iterator.hasNext()) {
					Object[] leftRecord = (Object[]) iterator.next();
					Object key = leftRecord[leftColumn];
					JoinedData rightJoinedData = 
							resultMap.putIfAbsent(key, new JoinedData(leftRecord, null));
					if( rightJoinedData != null ) {
						Object[] joinedRecord = new Object[joinedRecordLength];   
						System.arraycopy(leftRecord, 0, joinedRecord, 0, leftRecordLength);
						System.arraycopy(rightJoinedData.getRight(), 0, joinedRecord, leftRecordLength, rightRecordLength);
//						System.out.println("l2r joinedRecord: " + Arrays.toString(joinedRecord));					
						if( queue.offer(joinedRecord, 10, TimeUnit.SECONDS) ) {
							resultMap.remove(key);
						} else {
							throw new IllegalStateException("Can't insert record into the queue.");
						}
					}				
				}
				

				rightToLeftFuture.get();
				handleUnmatchedRecords();
				queue.offer(POISON);

				return null;
			}

			private void handleUnmatchedRecords() throws InterruptedException,
					AssertionError {
				Set<Entry<Object, JoinedData>> entrySet = resultMap.entrySet();
				for (Entry<Object, JoinedData> entry : entrySet) {
					JoinedData joinedData = entry.getValue();
					Object[] leftRecord = joinedData.getLeft();
					Object[] rightRecord = joinedData.getRight();
					if( leftRecord != null && rightRecord == null ) {
						Object[] joinedRecord = new Object[joinedRecordLength];   
						System.arraycopy(leftRecord, 0, joinedRecord, 0, leftRecordLength);
						if( !queue.offer(joinedRecord, 10, TimeUnit.SECONDS) ) {
							throw new IllegalStateException("Can't insert record into the queue.");
						}
					} else if (leftRecord == null && rightRecord != null) {
						LOG.log(Level.WARNING, "Right record: " + 
					         Arrays.toString(rightRecord) + " is unmatched.");
					} else {
						throw new AssertionError("Wrong logic.");
					}
				}
				resultMap.clear();
			}
		});
	}

	private void joinFromRightToLeft() {
		this.rightToLeftFuture = threadPool.submit(new Callable<Void>() {
			public Void call() throws Exception {
				Iterator<Object[]> iterator = rightDataSource.iterator();
				while (iterator.hasNext()) {
					Object[] rightRecord = (Object[]) iterator.next();
					Object key = rightRecord[rightColumn];
//					System.out.println("rightRecord: " + Arrays.toString(rightRecord));
					JoinedData leftJoinedData = resultMap.putIfAbsent(key, new JoinedData(null, rightRecord));
					if( leftJoinedData != null ) {
						Object[] joinedRecord = new Object[joinedRecordLength];   
						System.arraycopy(leftJoinedData.getLeft(), 0, joinedRecord, 0, leftRecordLength);
						System.arraycopy(rightRecord, 0, joinedRecord, leftRecordLength, rightRecordLength);
//						System.out.println("r2l joinedRecord: " + Arrays.toString(joinedRecord));											
						if( !queue.offer(joinedRecord, 10, TimeUnit.SECONDS) ) {
							throw new IllegalStateException("Can't insert record into the queue.");
						}
					}
				}
				
				return null;
			}
		});
	}

	public void begin() {
		threadPool = Executors.newFixedThreadPool(2);
		leftDataSource.begin();
		rightDataSource.begin();
		joinRows();
	}

	public void end() {		
		try {
			rightToLeftFuture.get();
			leftToRightFuture.get();
			
			leftDataSource.end();
			rightDataSource.end();

			threadPool.shutdownNow();
			resultMap.clear();
		} catch (InterruptedException e) {
			LOG.log(Level.SEVERE, "Interrupted by user.", e);
			Thread.currentThread().interrupt();
		} catch (ExecutionException e) {
			LOG.log(Level.SEVERE, "Exception during execution.", e);
			throw new IllegalStateException(e);
		}
	}
}
